package com.example.demo.mqserver.datacenter;

/**
 * Created with IntelliJ IDEA.
 * Description
 * User: 杰
 * Date: 2023 -07 -31
 * Time: 15:49
 */

import com.example.demo.common.BinaryTool;
import com.example.demo.common.MqException;
import com.example.demo.mqserver.core.MSGQueue;
import com.example.demo.mqserver.core.Message;
import jdk.internal.util.xml.impl.Input;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 通过这个类, 来针对硬盘上的消息进行管理
 */
public class MessageFileManager {
    // 定义内部类来表示该对列的统计信息
    static public class Stat{
        public int totalCount; // 总消息数量
        public int validCount; // 有效消息数量
    }

    public void init() {
        // 暂无初始化工作.
    }
    // 约定消息文件所在的目录和文件名
    // 这个方法, 用来获取到指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 这个方法用来获取该队列的消息数据文件路径
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    // 该方法用来获取该队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName) {
        // 由于当前的消息统计文件是文本文件, 可以直接使用 Scanner 来读取文件内容
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void writeStat(String queueName, Stat stat) {
        // 使用 PrintWrite 来写文件
        // OutputStream 打开文件, 默认情况下, 会直接把原文件清空.(新的数据覆盖了旧的数据)
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            // 让缓冲区中的数据刷新到硬盘之中
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 不存在, 就创建这个目录
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给消息统计文件, 设定初始值.
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName,stat);
    }

    // 删除队列的目录和文件
    // 队列中的文件都要被删除
    public void destroyQueueFiles(String queueName) throws IOException {
        // 先删除目录里面的文件, 再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            // 如果有任意一个删除失败, 都算整体删除失败.
            throw new IOException("删除队列目录和文件失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }

    // 检查队列的目录和文件是否存在
    // 当后续有生产者给 broker server 生产消息了, 这个消息就可能需要记录到文件上(取决于消息是否要持久化)
    public boolean checkFilesExits(String queueName) {
        // 判定队列的数据文件和统计文件是否都存在!
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    // 该方法是把一个新的消息放到对应的队列文件中.
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 检查文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName=" + queue.getName());
        }
        // 2. 把 Message 对象, 进行序列化, 转成二进制的字节数组.
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue) {
            // 3. 先获取到当前的队列数据文件的长度, 用这个来计算该 Message 对象的 offsetBeg 和 offsetEnd.
            //把新的 Message 数据, 写入到队列数据文件的末尾. 此时 Message 对象的 offsetBeg, 就是当前文件长度 + 4.
            //offsetEnd 就是当前文件长度 + 4 + message 自身长度.
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // 通过该方法(queueDataFile.length())就可以获取到文件的长度.  单位是字节.
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 4. 写入消息到数据文件. 注意 这里是追加写入到文件末尾.
            try (OutputStream outputStream = new FileOutputStream(queueDataFile,true)){
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 先写当前消息的长度, 占据 4个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.validCount += 1;
            stat.totalCount += 1;
            writeStat(queue.getName(),stat);
        }
    }

    // 删除消息
    // 进行逻辑删除, 将isValid 属性设置为 0.
    // 1. 先把文件中的这一段数据, 读出来, 还原回 Message 对象.
    // 2. 将 isValid 设置为 0.
    // 3. 把上述数据重新写会到文件中.
    // 此处这个参数中的 message 对象, 必须得包含有效的 offsetBeg 和 offsetEnd.
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")) {
                // 1. 先从文件中读取对应的 Message 数据.
                byte[] bufferSrc = new byte[(int)(message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 2. 把当前读出的二进制数据, 转回成 Message 对象.
                Message disMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 3. 把 isValid 设置为 0.
                disMessage.setIsValid((byte)0x0);
                // 此处不需要给参数的这个 message 的 isValid 设为0, 因为这个参数代表的是内存中管理的 Message 对象.马上就会被内存销毁了.
                // 4. 重写写入文件
                byte[] bufferDest = BinaryTool.toBytes(disMessage);
                // 重点注意 因为上述进行了 read 操作, 文件光标往后移动了. 因此还需要再调整光标的位置.
                randomAccessFile.seek(message.getOffsetBeg());
                // 这里的 write 会把之前的位置上的数据都覆盖掉
                randomAccessFile.write(bufferDest);
            }
            // 更新统计文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(),stat);
        }
    }

    // 从文件中, 读取出所有的消息内容, 加载到内存中
    // 在程序启动的时候, 进行调用.
    // 这里使用 LinkedList, 目的是为了头删操作.
    // 参数 使用 queueName 不需要 MSGqueue对象 是因为不涉及到加锁操作. 因为该方法是在程序启动时调用, 此时服务器还不能处理请求,
    // 因此不存在 线程安全的问题.
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)){
                // 使用该变量来记录当前文件光标.
                long currentOffset = 0;
                // 一个文件包含了很多信息, 此处势必要循环读取.
                while (true) {
                    // 1. 读取当前消息的长度, 这里的 readInt 可能会读到文件的末尾(EOF).
                    // readInt 方法读到文件末尾后, 会抛出 EOFException 异常.
                    int messageSize = dataInputStream.readInt();
                    // 2. 按照这个长度, 读取消息内容.
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        // 如果不匹配, 证明文件有问题, 格式错乱!
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }
                    // 3. 把读出的二进制数据, 反序列化回 Message 对象.
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 4. 判定一下看看这个消息对象是不是无效对象.
                    if (message.getIsValid() != 0x1) {
                        // 无效数据, 跳过, 但是 offset 还是需要更新的.
                        currentOffset += (4+messageSize);
                        continue;
                    }
                    // 5. 有效数据, 则需要把这个 Message 对象添加到链表中,
                    // 加入之前还需要填写 offsetBeg 和 offsetEnd
                    //  当下的这两个流对象都不能移动光标的位置, 因此就需要手动计算.
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4+messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 这个异常, 表示文件读到了异常, 是 readInt 抛出的异常.
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
        return messages;
    }

    // 检查当前是否要针对该队列的消息数据文件进行 GC
    public boolean checkGC(String queueName) {
        // 判定是否要GC,  根据总消息和有效消息.
        Stat stat = readStat(queueName);
        if ((stat.totalCount) > 2000 && ((double)stat.validCount / (double)stat.totalCount < 0.5)) {
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    // 通过这个方法来执行数据文件的垃圾回收操作.
    // 使用复制算法来完成.
    // 创建一个新的文件命名为 queue_data_new.txt.
    // 把之前的数据文件中的有效消息都读出来, 写到新的文件中.
    // 赋值完成之后, 删除旧的文件, 将新的文件名改回 queue_data.txt. 同时更新消息统计文件
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 进行 gc 的时候, 不允许其他线程对该队列进行文件的修改操作. 加锁
        synchronized (queue) {
            // 由于 gc 操作可能比较耗时, 此处统计一下执行消耗的时间.
            long gcBeg = System.currentTimeMillis();
            // 1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 正常情况下, 这个文件是不应该存在的, 如果存在, 证明上次 gc 出错了.程序已经崩溃.
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在! " +
                        "queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile+" + queueDataNewFile.getAbsolutePath());
            }
            // 2. 从旧的文件中, 读取出所有的有效消息对象.
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 3. 将有效消息, 写入新文件中.
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        // 先写四个字节消息的长度.
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            // 4. 删除旧的数据文件, 并且把新的文件进行重命名.
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("[MessageFileManager] 删除就的数据文件失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            // 进行重名名操作. queue_data_new.txt -> queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath() +
                        ",queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            // 5. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕! queueName=" + queue.getName() +
                    ", time=" + (gcEnd-gcBeg) + "ms");
        }
    }
}



























